package com.xinshujia.listener;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import java.util.UUID;

import static com.xinshujia.config.mqtt.MqttConstant.CHANNEL_NAME_IN;

/**
 * *************************************
 * *   @author: 野狼上校               **
 * *   @since: 2024-07-25 17:24       **
 * *   @version: 1.0                  **
 * *************************************
 * <p>
 * 监听订阅的主题
 */
@Configuration
public class MqttListener {
    // 客户端ID
    @Value("${spring.mqtt.clientId}")
    private String clientId;
    // 订阅主题
    @Value("${spring.mqtt.topic.subscribe}")
    private String subscribeTopic;

    // MQTT客户端的连接工厂
    private final MqttPahoClientFactory mqttClientFactory;

    public MqttListener(MqttPahoClientFactory mqttClientFactory) {
        this.mqttClientFactory = mqttClientFactory;
    }


    /**
     * MQTT消息通道 - 入站（消费者）
     */
    @Bean(name = CHANNEL_NAME_IN)
    public MessageChannel mqttInboundChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT消息订阅绑定 - 入站（消费者）
     */
    @Bean
    public MessageProducer inbound() {
        //配置订阅端MqttPahoMessageDrivenChannelAdapter
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId + UUID.randomUUID().toString().substring(0,8), mqttClientFactory, subscribeTopic);
        //设置超时时间
        adapter.setCompletionTimeout(5000);
        //设置默认的消息转换类
        adapter.setConverter(new DefaultPahoMessageConverter());
        //设置qos级别
        adapter.setQos(1);
        //设置入站管道
        adapter.setOutputChannel(mqttInboundChannel());
        return adapter;
    }

    /**
     * MQTT消息监听器（消费者）
     * MessageHandler: org.springframework:spring-messaging
     */
    @Bean
    @ServiceActivator(inputChannel = CHANNEL_NAME_IN)
    public MessageHandler handlerTest() {
        return message -> {
            try {
//                String json = new Gson().toJson(message);
//                System.out.println(json);
                String payload = message.getPayload().toString();
                System.out.println(payload);
            } catch (MessagingException e) {
                e.printStackTrace();
            }
        };
    }
}
